// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: fengyang

#include "coin2/exchange/dydx_futures/feed_v3/processor.h"

#include <string_view>

#include <rapidjson/error/en.h>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/dydx_futures/feed_v3/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::dydx_futures::feed_v3::impl {
void DydxFuturesFeedProcessor::OnTopicRecord(const TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error("JSON parse error: "s + GetParseError_En(document.GetParseError()));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.HasMember("channel")) {
    const std::string channel = document["channel"].GetString();
    if (!CheckChannelFilter(channel)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    const std::string native_symbol = document["id"].GetString();
    if (channel == "v3_trades") {
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }

      Trade* trade = upd_mgr_.GetTradeMutable();
      trade->Clear();

      int index = 0;
      int last = document["contents"]["trades"].Size();
      for (const auto& trade_json : document["contents"]["trades"].GetArray()) {
        DydxFuturesFeedUpdateParser::ParseTrade(trade_json, trade);
        index++;
        upd_mgr_.PublishTrade(native_symbol, index == last);
      }
    } else if (channel == "v3_orderbook") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        DydxFuturesFeedUpdateParser::ParseBook(book, document, packet.timestamp);
        upd_mgr_.PublishBook();
      }
    } else {
      THROW() << "[DydxFutures] unknown channel: " << channel;
    }
  } else if (document.HasMember("markets")) {
    if (!CheckChannelFilter("markets")) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    for (const auto& market : document["markets"].GetObject()) {
      const auto& native_symbol = market.name.GetString();
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        continue;
      }
      const auto& data = market.value;

      auto* index = upd_mgr_.GetIndexMutable();
      index->Clear();
      index->timestamp = packet.timestamp;
      index->price = stod(data["indexPrice"].GetString());
      upd_mgr_.PublishIndex(native_symbol);

      auto* open_interest = upd_mgr_.GetOpenInterestMutable();
      open_interest->timestamp = packet.timestamp;
      open_interest->open_interest_qty = stod(data["openInterest"].GetString());
      upd_mgr_.PublishOpenInterest(native_symbol);

      auto* funding_rate = upd_mgr_.GetFundingRateMutable();
      funding_rate->Clear();
      funding_rate->timestamp = packet.timestamp;
      funding_rate->funding_rate = stod(data["nextFundingRate"].GetString());
      funding_rate->funding_time =
          TimestampFromIso8601(data["nextFundingAt"].GetString());
      upd_mgr_.PublishFundingRate(native_symbol);
    }
  } else if (document.HasMember("url")) {
    const std::string url = document["url"].GetString();
    if (url.find("candles") != std::string::npos) {
      auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
      std::string resolution_str = "resolution=" + parameters["resolution"];
      if (!CheckChannelFilter(resolution_str)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }

      size_t start = url.find_last_of('/') + 1;
      size_t end = url.find_last_of('?');
      const std::string& native_symbol = url.substr(start, end - start);
      auto interval = DydxFuturesFeedUpdateParser::ParseKlineInterval(parameters.at("resolution"));
      auto* kline = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (kline != nullptr) {
        bool to_publish = false;
        // query is split up several requests
        int page = -1;
        if (parameters.count("page") > 0) {
          page = std::stod(parameters["page"]);
        }
        if ((page == 0 ) || (page == -1 && document["data"]["candles"].Size() > 10) ) {
          // first long query
          kline->SnapshotClear();
        } else {
          kline->UpdateClear();
        }

        const auto& items = document["data"]["candles"].GetArray();
        for (int i = items.Size() - 1; i >= 0; i--) {
          const auto& kline_data = items[i];
          bool freezed = DydxFuturesFeedUpdateParser::Parse1Kline(
              kline_data, kline, packet.timestamp, interval);
          to_publish = to_publish || freezed;
        }

        if (to_publish) {
          upd_mgr_.PublishKline(native_symbol);
        }
      }
    }
  } else {
    LOG(INFO) << "[DydxFutures] meta msg for " << packet.topic_name << ": "
        << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}  // class DydxFuturesFeedProcessor
}  // namespace coin2::exchange::dydx_futures::feed_v3::impl
